/**
  * @brief project LoadBanlancer,
  * @file LoadBanlancer.cpp
  * @author Liu Shuo
  * @date 2011.5.15
  */

#include "main.h"
#include "LoadBanlancer.h"

/**
  * @class LoadBanlancer
  * @brief 用于创建线程
  * @param [in] arg: NULL
  * @return NULL
  */
void *thread_recvhearbeat(void *arg)
{
    return ((LoadBanlancer *)arg)->receiveheartbeat(NULL);
}

/**
  * @class LoadBanlancer
  * @brief 用于创建线程
  * @param [in] arg: NULL
  * @return NULL
  */
void *thread_recvmessage(void *arg)
{
    return ((LoadBanlancer *)arg)->receiveMessage(NULL);
}

/**
  * @class LoadBanlancer
  * @brief 用于创建线程
  * @param [in] arg: NULL
  * @return NULL
  */
void *thread_changeStatus(void *arg)
{
    return ((LoadBanlancer *)arg)->changeStatus(NULL);
}

/**
  * @brief 用于创建线程
  * @param [in] arg: struct request Args *
  * @return NULL
  */
void *thread_handleRequest(void *arg)
{
    //cout << "thread_handleRequest" << endl;
    Pthread_detach(pthread_self());
    struct requestArgs *ptr = (struct requestArgs *)arg;
    //ptr->topNode->addThreadNum();
    ptr->LB->handleRequest(ptr->connfd, &ptr->cliaddr);
    free(ptr);
    return NULL;
}

/**
  * @class TopNode
  * @brief 用于创建线程
  * @param [in] arg: NULL
  * @return NULL
  */
void *thread_handleMesg(void *arg)
{
    Pthread_detach(pthread_self());
    struct requestArgs *ptr = (struct requestArgs *)arg;
    ptr->LB->handleMesg(ptr->connfd, &ptr->cliaddr);
    free(arg);
    return (void *)NULL;
}

/**
  * @class LoadBanlancer
  * @brief 构造函数
  */
LoadBanlancer::LoadBanlancer()
{
    Pthread_mutex_init(&m_nextNodeLock, NULL);
    m_nextNode = 0;
    m_rsNum = 0;
}

/**
  * @class LoadBanlancer
  * @brief 析构函数
  */
LoadBanlancer::~LoadBanlancer()
{
    pthread_mutex_destroy(&m_nextNodeLock);
}

/**
  * @class TopNode
  * @brief initialize the LoadBanlancer and run
  */
void LoadBanlancer::run()
{
    Pthread_create(&m_heatbeatPthread, NULL, thread_recvhearbeat, (void *)this);
    Pthread_create(&m_messagePthread, NULL, thread_recvmessage, (void *)this);
    Pthread_create(&m_statusPthread, NULL, thread_changeStatus, (void *)this);
    serve();
}

/**
  * @class LoadBanlancer
  * @brief 对外提供服务
  */
void LoadBanlancer::serve()
{
    int listenfd = Socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in serveraddr;
    int connfd;
    socklen_t clilen;
    struct requestArgs *args;
    pthread_t tid;

    bzero(&serveraddr, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = htons(INADDR_ANY);
    serveraddr.sin_port = htons(m_servicePort);

    Bind(listenfd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
    Listen(listenfd, MAXCONNUM);

    while (1) {
        args = (struct requestArgs *)Malloc(sizeof(struct requestArgs));
        clilen = sizeof(args->cliaddr);
        connfd = Accept(listenfd, (struct sockaddr *)&(args->cliaddr), &clilen);
        args->LB = this;
        args->connfd = connfd;
        Pthread_create(&tid, NULL, thread_handleRequest, (void *)args);
    }
}

void *readRS(void *args)
{
    Pthread_detach(pthread_self());
    int len;
    char buf[MAXLINE];
    s_sock *arg = (s_sock *)args;
    while ((len = read(arg->sockfd, buf, MAXLINE)) > 0) {
        //将数据发送给客户端
        cout << buf << endl;
        Writen(arg->connfd, buf, len);
        bzero(buf, MAXLINE);
    }
    return NULL;
}

/**
  * @class LoadBanlancer
  * @brief handle the request from client
  * @param [in] connfd: the socket ID of client
  * @param [in] cliaddr: 客户端地址结构
  */
void *LoadBanlancer::handleRequest(int connfd, const struct sockaddr_in *cliaddr)
{
    ssize_t len;
    char buf[MAXLINE];
    Node *node;
    int sockfd;
    pthread_t tid;
    s_sock arg;

    if (m_policy == 0) {
        node = getNextNode();
    }
    else {
        node = getNextNodeByIP(cliaddr->sin_addr.s_addr);
    }
    sockfd = node->connectServer();

    arg.connfd = connfd;
    arg.sockfd = sockfd;
    Pthread_create(&tid, NULL, readRS, &arg);
    while ((len = read(connfd, buf, MAXLINE)) > 0) {
        write(sockfd, buf, len);
    }
    pthread_cancel(tid);
    node->closeServer(sockfd);
    Close(connfd);
    cout << "service end" << endl;
    return NULL;
}


/**
  * @class LoadBanlancer
  * @brief 接收来自RS的消息
  * @param arg is NULL
  * @return NULL
  */
void *LoadBanlancer::receiveMessage(void *arg)
{
    int socketID, connfd;
    socklen_t clilen;
    socketID = Socket(AF_INET, SOCK_STREAM, 0);
    pthread_t tid;
    struct requestArgs* args;
    struct sockaddr_in serveraddr;

    bzero(&serveraddr, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
    serveraddr.sin_port = htons(m_messagePort);

    Bind(socketID, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
    Listen(socketID, 10);
    while (1)
    {
        args = (struct requestArgs *)Malloc(sizeof(struct requestArgs));
        clilen = sizeof(args->cliaddr);
        connfd = Accept(socketID, (struct sockaddr *)&(args->cliaddr), &clilen);
        args->LB = this;
        args->connfd = connfd;
        Pthread_create(&tid, NULL, thread_handleMesg, args);
    }
    return (void *)NULL;
}

/**
  * @class LoadBanlancer
  * @brief 处理接收的消息
  * @param [in] connfd: the socket ID of client
  * @param [in] cliaddr: 客户端地址结构
  */
void *LoadBanlancer::handleMesg(int connfd, const struct sockaddr_in *cliaddr)
{
    char buf[MAXLINE] = {0};
    string mesg = "";
    int i;

    while (read(connfd, buf, MAXLINE) > 0)
    {
        cout << buf << endl;
        mesg.append(buf);
        bzero(buf, MAXLINE);
    }
    switch (mesg[0]) {
        case 'R': { //注册新节点, 故障节点恢复
            i = 2;
            string nodeID("");
            string strPort("");
            string strProperty("1");
            while (mesg[i] != ',')
            {
                nodeID.push_back(mesg[i]);
                i++;
            }
            i++;
            for (int j = 0; j < nodeVector.size(); j++)
            {
                if (nodeVector[j]->getID() == nodeID) {
                    printf("节点%s恢复\n", nodeID.c_str());
                    nodeVector[j]->setstatus(LIVE);
                    Close(connfd);
                    return NULL;
                }
            }
            while (mesg[i] != '\n')
            {
                strPort.push_back(mesg[i]);
                i++;
            }
#ifdef DEBUG
            cout << "register new node, node ID: "<< nodeID << endl;
            cout << "node service port: " << strPort << endl;
#endif
            Node *newNode = new Node(nodeID, strPort, cliaddr->sin_addr.s_addr, atoi(strProperty.c_str()));
            newNode->setstatus(LIVE);
            nodeVector.push_back(newNode);
            m_rsNum++;
            break;
        }
        default:
            break;
    }
    Close(connfd);
    return NULL;
}


/**
  * @class LoadBanlancer
  * @brief 初始化负载均衡服务器
  * @param [in] arg: NULL
  * @return NULL
  */
void *LoadBanlancer::receiveheartbeat(void *arg)
{
#ifdef DEBUG
    cout << "execute receiveheartbeat()" << endl;
#endif
    int socketID = Socket(AF_INET, SOCK_DGRAM, 0);

    struct sockaddr_in serveraddr, clientaddr;
    bzero(&serveraddr, sizeof(serveraddr));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
    serveraddr.sin_port = htons(m_heartbeatPort);

    Bind(socketID, (struct sockaddr *) &serveraddr, sizeof(serveraddr));
    char message[MAXLINE];
    socklen_t len;
    while (1)
    {
        Recvfrom(socketID, message, MAXLINE, 0, (struct sockaddr *) &clientaddr, &len);
#ifdef DEBUG
        //cout << message << endl;
#endif
        int i = 2;
        string nodeID("");
        while (message[i] != 0)
        {
            nodeID.push_back(message[i]);
            i++;
        }
        for (vector<Node *>::size_type i = 0; i != nodeVector.size(); i++)
        {
            if (nodeID == nodeVector[i]->getID())
            {
                nodeVector[i]->setheartBeatTime(time(NULL));
                nodeVector[i]->setstatus(LIVE);
#ifdef DEBUG
                //cout << nodeVector[i]->getID() << " heart beat time: "
                 //    << nodeVector[i]->getheartBeatTime() << endl;
#endif
                break;
            }
        }
    }
    return (void *)NULL;
}


/**
  * @class LoadBanlancer
  * @brief 检查节点状态
  * @param arg: NULL
  * @return NULL
  */
void *LoadBanlancer::changeStatus(void *arg)
{
    while (1) {
        sleep(1);
        for (vector<Node *>::size_type i = 0; i != nodeVector.size(); i++) {
            if (nodeVector[i]->getstatus() == LIVE) {
                if ((time(NULL) - nodeVector[i]->getheartBeatTime()) > 2) {
                    cout << "node " << nodeVector[i]->getID() << " is dead!" << endl;
                    nodeVector[i]->setstatus(DEAD);
                }
            }
        }
    }
    return (void *)NULL;
}

/**
  * @class LoadBanlancer
  * @brief 获得下一个可用的节点指针
  * @return 节点指针
  */
Node *LoadBanlancer::getNextNode()
{
    Pthread_mutex_lock(&m_nextNodeLock);
    while (1) {
        if (m_nextNode >= nodeVector.size()) {
            m_nextNode = 0;
        }
        if (nodeVector[m_nextNode]->getstatus() == LIVE) {
            Node *result = nodeVector[m_nextNode];
            m_nextNode++;
			Pthread_mutex_unlock(&m_nextNodeLock);
            return result;
        }
        m_nextNode++;
    }
}

/**
  * @class LoadBanlancer
  * @brief 通过ip hash,获得下一个可用的节点指针
  * @param [in] ip: 客户端ip
  * @return 节点指针
  */
Node *LoadBanlancer::getNextNodeByIP(uint32_t ip)
{
    uint32_t index;
    Node *result;
    if (m_rsNum != 0)
        index = ip % m_rsNum;
    else {
        return NULL;
    }
    if (nodeVector[index]->getstatus() == LIVE) {
        result = nodeVector[index];
        return result;
    }
    else {
        return getNextNode();
    }
    return getNextNode();
}

/**
  * @class LoadBanlancer
  * @brief 读取配置文件
  * @param [in] 配置文件地址
  */
void LoadBanlancer::readConf(char *filename)
{
    FILE *fp = Fopen(filename, "r");
    int num = 0;
    char buf[MAXLINE];

    while (Fgets(buf, MAXLINE, fp) != NULL)
    {
        if (buf[0] != '#')
        {
            buf[strlen(buf) - 1] = 0;
            switch (num)
            {
            case 0:
                m_heartbeatPort = atoi(buf);
                break;
            case 1:
                m_messagePort = atoi(buf);
                break;
            case 2:
                m_servicePort = atoi(buf);
                break;
            case 3:
                m_policy = atoi(buf);
                break;
            default:
                break;
            }
            num++;
        }
    }
#ifdef DEBUG
    cout << "heartbeat port: " << m_heartbeatPort << endl;
    cout << "message port: " << m_messagePort << endl;
    cout << "service port: " << m_servicePort << endl;
#endif
}
